package mrt.cse.gajanidu.Manager;

import org.apache.zookeeper.*;
import org.apache.zookeeper.data.Stat;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CountDownLatch;
/*
This class connect to the given server , run a check for lead
manager periodically when it is triggered node change in parent
node
*/

public abstract class NodeWatcher implements Watcher {

    protected ZooKeeper zk;
    private final int SESSION_TIME_OUT = 5000;
    protected final String defPath = "/managers";
    protected boolean isLeader = false;
    protected String coordinatorPath;
    protected String coordinatorName;
    List<String> managersList = new ArrayList<String>();
    private long coordinatorId = 0;
    public String managerName;
    protected boolean expired = false;

    private CountDownLatch connectedSignal = new CountDownLatch(1);


    /**
     * Connects to the ZooKeeper Service
     *
     * @param host address of the ZooKeeper Server
     */
    public void connectToServer(String host) {
        try {
            zk = new ZooKeeper(host, SESSION_TIME_OUT, this);
            connectedSignal.await();
        } catch (IOException e) {
            System.out.println(e.toString());
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    /**
     * checks for the existence of the persistent znode,
     * If it is existing either it will set a watcher or not depending on the addWatch
     *
     * @param path     path of the node
     * @param addWatch true if a watcher is needed, else if false
     * @return Stat
     */
    protected Stat isNodeExist(String path, boolean addWatch) {

        if (path != null) {
            try {
                if (addWatch) {
                    return zk.exists(path, this);
                } else {
                    return zk.exists(path, false);
                }
            } catch (KeeperException e) {
                e.printStackTrace();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        return null;
    }

    /**
     * Creates a ZooKeeper Node for the Current manager under the Main node
     *
     * @param path path to the node to be created
     * @param data data to be saved in the node
     */
    protected void createZnode(String path, String data) {
        try {
            zk.create(path, data.getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
            System.out.println("INFO :  Successfully created the Node in " + path);
        } catch (KeeperException e) {
            System.out.println(e.toString());
        } catch (InterruptedException e) {
            System.out.println(e.toString());
        }
    }

    /**
     * Returns the set of children under the main node
     *
     * @return A list containing the names of children of the main node
     */
    protected List<String> getChildrenOfMainNode() {
        try {
            return zk.getChildren(defPath, false);
        } catch (KeeperException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return new ArrayList<String>();
    }

    /**
     * Assigns him self as the coordinator of the Manager Cloud
     *
     * @param path          path of the Manager
     * @param coordinatorId ID of the Manager
     */
    protected void assignCoordinatorMySelf(String path, String coordinatorId) {
        this.coordinatorId = Long.valueOf(coordinatorId);
        this.coordinatorPath = path;
        isLeader = true;
        writeCoordinatorData(managerName + String.valueOf(coordinatorId));
        System.out.println("INFO :  No Coordinator in the cloud");
        System.out.println("INFO :  " + managerName + " is elected as the Coordinator");
    }


    /**
     * Writes the name of the Leader into the Parent Node
     *
     * @param data
     */
    private void writeCoordinatorData(String data) {
        byte[] bytes = data.getBytes();
        try {
            zk.setData(defPath, bytes, -1);
        } catch (KeeperException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    /**
     * Checks the status of the Coordinator in the system
     * If this is the only Manager in the cloud, it will be assigned as the Coordinator or else
     * coordinator will be elected
     */
    protected void checkCoordinator() {
        List<String> childList = getChildrenOfMainNode();
        if (childList.size() == 1) {
            assignCoordinatorMySelf(childList.get(0), getZidOfNode(childList.get(0)));
        } else {
            if (!isLeader) {
                electCoordinator(childList);
            }
        }
        addWatcher();
    }

    /**
     * Adds a watcher to the Main node so that changes can be observed.
     */
    protected void addWatcher() {
        try {
            zk.getChildren(defPath, true);
        } catch (KeeperException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

    }

    /**
     * This method will be executed when there are previous managers in the cloud which implies
     * there is a Coordinator. Coordinator name is saved in the parent node. It will be retrieved
     * in this method
     */
    protected void processForCoordinator() {
        try {
            byte[] bytes = zk.getData(defPath, false, null);
            String name = new String(bytes);
            coordinatorPath = name;
            coordinatorName = name.substring(0, name.length() - 10);
            System.out.println("INFO :  A Coordinator is present in the cloud");
            System.out.println("INFO :  The Coordinator is : " + this.coordinatorName);
        } catch (KeeperException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

    }


    protected boolean electCoordinator(List<String> newList) {
        //TODO- this process crash when sequence id is higher than a long value do something for that
        System.out.println(managerName + " :    Starting to elect coordinator");
        int count = 0;
        long tempId;
        long tempCoordinatorId = 0;
        String tempCoordinatorPath = "";
        for (String str : newList) {
            tempId = Long.valueOf(getZidOfNode(str));
            if (count == 0) {
                tempCoordinatorId = tempId;
                count++;
            }
            if (tempId != 0) {
                if (tempId <= tempCoordinatorId) {
                    tempCoordinatorId = tempId;
                    tempCoordinatorPath = str;
                }
            }
        }
        if (tempCoordinatorId > 0 && !tempCoordinatorPath.equalsIgnoreCase("")) {
            coordinatorId = tempCoordinatorId;
            coordinatorPath = tempCoordinatorPath;
            System.out.println(managerName + " :    Coordinator path is : " + coordinatorPath + "   " +
                    "   Coordinator name : " + coordinatorPath.substring(0,
                    coordinatorPath.length() - 10));
            writeCoordinatorData(coordinatorPath);
            return true;
        } else {
            return false;
        }
    }


    protected String getZidOfNode(String path) {
        int len;
        if (path != null) {
            len = path.length();
            if (len > 10) {
                return path.substring(len - 10, len);
            }
        }
        return "";
    }

    @Override
    public void process(WatchedEvent event) {
        if (event.getType() == Event.EventType.None) {
            switch (event.getState()) {
                case SyncConnected:
                    System.out.println("INFO :  Connected to the ZooKeeper Service");
                    connectedSignal.countDown();
                    break;
                case Expired:
                    System.out.println("ERROR :  Connection to the ZooKeeper service has lost");
                    expired = true;
            }
        } else if (event.getType() == Event.EventType.NodeChildrenChanged) {
            System.out.println(managerName + " :    Children change in - " + event.getPath());
            checkCoordinator();
        }
    }

}
